package com.dd.keel.core.framework.server;

import org.apache.mina.core.service.IoHandlerAdapter;
import org.apache.mina.core.session.IdleStatus;
import org.apache.mina.core.session.IoSession;
import org.apache.mina.transport.socket.SocketSessionConfig;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * 服务器端消息接收器(带心跳功能)
 * @author Kevin.XU
 *
 */
public class ServerReceiverHandler extends IoHandlerAdapter{

	final static private Logger LOGGER = LoggerFactory.getLogger(ServerReceiverHandler.class);
	
	private ServerReqObjQueue serverReqObjQueue;   /**请求消息队列**/
	private int idle_seconds = 120;        /**最大空闲时间，默认值120**/
	private int max_connection_num = 500;  /**最大连接数，默认值500*/
	private ServerHeartbeatHandler heartbeatHandler;  /**心跳处理器*/
	
	public ServerReceiverHandler(){}
	
    public ServerReqObjQueue getServerReqObjQueue() {
		return serverReqObjQueue;
	}

	public void setServerReqObjQueue(ServerReqObjQueue serverReqObjQueue) {
		this.serverReqObjQueue = serverReqObjQueue;
	}

	public int getIdle_seconds() {
		return idle_seconds;
	}

	public void setIdle_seconds(int idle_seconds) {
		if(idle_seconds>0) this.idle_seconds = idle_seconds;
	}

	public int getMax_connection_num() {
		return max_connection_num;
	}

	public void setMax_connection_num(int max_connection_num) {
		if(max_connection_num>0) this.max_connection_num = max_connection_num;
	}

	public ServerHeartbeatHandler getHeartbeatHandler() {
		return heartbeatHandler;
	}

	public void setHeartbeatHandler(ServerHeartbeatHandler heartbeatHandler) {
		this.heartbeatHandler = heartbeatHandler;
	}

	public void sessionCreated(IoSession session) 
    {
    	LOGGER.info("%%%%%remote client {} connected, total current session count is {}", session.getRemoteAddress(), session.getService().getManagedSessionCount());
    	if(session.getService().getManagedSessionCount()>max_connection_num)
    	{
    		LOGGER.warn("%%%%%exceed max connection count {},total current session count is {}, so close this connection",max_connection_num,session.getService().getManagedSessionCount());
    		LOGGER.info("%%%%%close remote client {}", session.getRemoteAddress() );
    		session.close(false);
    	}
    }
    
    public void sessionClosed(IoSession session) throws Exception 
    {
        // Print out total number of bytes read from the remote peer.
    	LOGGER.info("session {} closed, total read {} bytes", session.getRemoteAddress(), session.getReadBytes());
    	super.sessionClosed(session);
    }
    
    public void sessionOpened(IoSession session) 
    {
    	LOGGER.info("%%%%%remote client {} opened.", session.getRemoteAddress() );
        LOGGER.info("%%%%%total current session count is {}",session.getService().getManagedSessionCount());
        ((SocketSessionConfig) session.getConfig()).setReceiveBufferSize(37*1024);
        ((SocketSessionConfig) session.getConfig()).setSendBufferSize(25*1024);
        ((SocketSessionConfig) session.getConfig()).setKeepAlive(false);
        ((SocketSessionConfig) session.getConfig()).setBothIdleTime(idle_seconds);
        ((SocketSessionConfig) session.getConfig()).setTcpNoDelay(true);
        ((SocketSessionConfig) session.getConfig()).setReuseAddress(true);
        LOGGER.trace("Receive buffer is {}", ((SocketSessionConfig) session.getConfig()).getReceiveBufferSize() );
        LOGGER.trace("Send buffer is {}", ((SocketSessionConfig) session.getConfig()).getSendBufferSize() );
    }
    
    public void sessionIdle(IoSession session, IdleStatus status) 
    {
    	int sessionIdleCount = session.getIdleCount(IdleStatus.BOTH_IDLE);
    	LOGGER.info("*** IDLE # session {}, idle {} ***", session.getRemoteAddress(), sessionIdleCount );
    	if(sessionIdleCount>3)
    	{
    		LOGGER.warn("*** close {}, because its idlecount {} > 3 ***", session.getRemoteAddress(), sessionIdleCount );
    		session.close(false);
    	}
    }
    
    public void exceptionCaught(IoSession session, Throwable cause) 
    {
    	LOGGER.warn("session {} : cause is {}.", session.getRemoteAddress(), cause.getClass().getName() );
    	LOGGER.error(cause.getMessage(), cause);
    	if(cause.getClass().getName().equals("java.lang.OutOfMemoryError"))
    	{
    		LOGGER.warn("heap memory is not enough, close connection!!!");
    		session.close(false);
    	}
    	else
    	{
    		LOGGER.warn("session occured exception, so close it.");
            session.close(false);
    	}
    }
    
    public void messageReceived(IoSession session, Object message) throws Exception 
    {
        LOGGER.trace("***receive one message from {} : {}", session.getRemoteAddress(), message );
        if(getHeartbeatHandler()!=null && getHeartbeatHandler().isHeartBeat(message))
        {
        	LOGGER.info("~~~heartbeat request from {} : {}", session.getRemoteAddress(), message );
        	Object hbResponse = getHeartbeatHandler().produceHeartbeatResponse(message);
        	session.write(hbResponse);
        	LOGGER.info("~~~heartbeat response to {} : {}", session.getRemoteAddress(), hbResponse );        	
        }
        else
        {
        	serverReqObjQueue.putReq(new ServerTaskObj(session,message));
        }
    }
    
    public void messageSent(IoSession session, Object message) throws Exception 
    {
    	LOGGER.trace("session {} send message.", session.getRemoteAddress() );
        super.messageSent(session, message);
    }
    
}
